// Copyright 2024 Xiguan Inc
// Author: xiguan
// Email: xiguan.teng@qq.com
// Created on 2024/8/8
// This is ...
//

#include "master.h"

#include <absl/flags/flag.h>
#include <absl/flags/parse.h>
#include <absl/strings/str_format.h>
#include <grpcpp/ext/proto_server_reflection_plugin.h>
#include <grpcpp/grpcpp.h>
#include <grpcpp/health_check_service_interface.h>

#include <random>
#include <thread>

ABSL_FLAG(uint16_t, client_port, 23202,
          "Server listened client port for the service");

ABSL_FLAG(uint16_t, master_port, 23222,
          "Server listened master port for the service");

ABSL_FLAG(uint32_t, reduce_nums, 3, "The num of reduce task");

static uint32_t ihash(std::string filename) {
  // 创建一个随机数生成器
  std::random_device rd;  // 非确定性的随机数生成器
  std::mt19937 gen(rd());  // 使用Mersenne Twister 19937算法的随机数生成器

  std::uniform_int_distribution<> dis(110, 10086);
  int                             ans = dis(gen);

  for (auto x : filename) {
    ans += x;
    ans %= absl::GetFlag(FLAGS_reduce_nums);
  }
  return ans;
}

void Master::push_map_task(std::string &filename) {
  map_task_.push_back(filename);
  work_status_.set_master_status(mapreduce::Status::MAPPING);
}

void MapReduceImpl::assign_work(mapreduce::WorkType *ans) {
  if (master_.work_status_.master_status() == mapreduce::Status::MAPPING) {
    auto first = master_.map_task_.front();
    master_.prosessing_task_.push_back({"map", first});
    ans->set_type(mapreduce::Status::MAPPING);
    ans->set_map_task(first);
    master_.map_task_.erase(master_.map_task_.begin());
  }
}

grpc::Status MapReduceImpl::ask4task(grpc::ServerContext    *context,
                                     const mapreduce::Empty *request,
                                     mapreduce::WorkType    *response) {
  response->Clear();
  switch (master_.work_status_.master_status()) {
    case mapreduce::Status::WAITING:
      response->set_type(mapreduce::Status::WAITING);
      break;
    case mapreduce::Status::MAPPING:
    case mapreduce::Status::REDUCING:
      assign_work(response);
  }
  return grpc::Status::OK;
}

void run_master_server(uint16_t port, Master &m) {
  grpc::ServerBuilder server_builder;
  auto          master_server_address = absl::StrFormat("0.0.0.0:%d", port);
  MapReduceImpl map_reduce_service(m);
  server_builder.AddListeningPort(master_server_address,
                                  grpc::InsecureServerCredentials());
  server_builder.RegisterService(&map_reduce_service);
  std::unique_ptr<grpc::Server> master_server(server_builder.BuildAndStart());

  master_server->Wait();
}

grpc::Status CountWordsImpl::count_files_words(
    grpc::ServerContext                   *context,
    const mapreduce_client::ClientRequest *request,
    mapreduce_client::ClientReply         *response) {
  int len = request->files_size();
  for (int idx = 0; idx < len; ++idx) {
    auto st = request->files(idx);
    master_.push_map_task(st);
  }
  std::cout << "Begin to mapping task." << std::endl;

  response->add_outfile("hello.");
  response->add_outfile("nmlgb.");
  return grpc::Status::OK;
}

void run_client_server(uint16_t port, Master &m) {
  grpc::ServerBuilder server_builder;
  auto                server_address = absl::StrFormat("0.0.0.0:%d", port);
  CountWordsImpl      client_service(std::forward<Master>(m));
  server_builder.AddListeningPort(server_address,
                                  grpc::InsecureServerCredentials());
  server_builder.RegisterService(&client_service);

  std::unique_ptr<grpc::Server> client_server(server_builder.BuildAndStart());

  client_server->Wait();
}

int main(int argc, char *argv[]) {
  absl::ParseCommandLine(argc, argv);
  grpc::EnableDefaultHealthCheckService(true);
  grpc::reflection::InitProtoReflectionServerBuilderPlugin();
  Master master;

  auto th_client = std::thread(
      run_client_server, absl::GetFlag(FLAGS_client_port), std::ref(master));
  auto th_master = std::thread(
      run_master_server, absl::GetFlag(FLAGS_master_port), std::ref(master));
  th_client.detach();
  th_master.detach();

  while (true) {
    continue;
  }
  return 0;
}